"""
Case Type   : Logical_Replication
Case Name   : 非初始用户创建、开启和删除逻辑复制槽，验证-W参数
Create At   : 2024/1/5
Owner       : xiaqi14
Description :
    1.修改参数
    2.创建sysadmin用户，配置用户白名单
    3.非初始用户不使用-w参数创建逻辑复制槽
    4.非初始用户使用-w参数创建逻辑复制槽
    5.非初始用户使用-w参数开启逻辑复制槽
    6.非初始用户使用-w参数删除逻辑复制槽
    7.非初始用户不使用-w参数删除逻辑复制槽
    8.删除用户
    9.恢复参数，恢复pg_hba.conf文件
Expect      :
    1.修改参数成功
    2.创建用户成功，配置白名单成功
    3.创建成功
    4.创建失败，合理报错
    5.开启失败，合理报错
    6.删除失败，合理报错
    7.删除成功
    8.删除成功
    9.恢复成功
History     :
"""
import os
import unittest
from yat.test import Node
from yat.test import macro
from testcase.utils.Logger import Logger
from testcase.utils.Common import Common
from testcase.utils.CommonSH import CommonSH
from testcase.utils.Constant import Constant
from testcase.utils.ComThread import ComThread


class LogicReplication(unittest.TestCase):
    def setUp(self):
        self.logger = Logger()
        self.logger.info(f"-----{os.path.basename(__file__)} start-----")
        self.primary_node = Node('PrimaryDbUser')
        self.sh_primary = CommonSH('PrimaryDbUser')
        self.constant = Constant()
        self.common = Common()
        self.user_name = 'u_logical_replication_0068'
        self.slot_name_1 = 'slot_logical_replication_0068_1'
        self.slot_name_2 = 'slot_logical_replication_0068_2'
        self.wrong_password = 'p_logical_replication_0068'
        self.config_wal_level = 'wal_level'
        self.wal_level_default = self.common.show_param(self.config_wal_level)

    def test_logical_replication(self):
        text = "-----step1:修改参数;expect:修改参数成功-----"
        self.logger.info(text)
        if 'logical' != self.wal_level_default:
            guc_result = self.sh_primary.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_wal_level}=logical')
            self.assertTrue(guc_result, '执行失败' + text)
            restart_msg = self.sh_primary.restart_db_cluster()
            self.logger.info(restart_msg)
            self.assertTrue(restart_msg, '重启失败')
        show_msg = self.common.show_param(self.config_wal_level)
        self.assertEquals(show_msg, 'logical', '执行失败' + text)

        text = "-----step2:创建sysadmin用户，配置用户白名单，" \
               "expect:创建用户成功，配置白名单成功-----"
        self.logger.info(text)
        sql_cmd = f'''drop user if exists {self.user_name};
            create user {self.user_name} sysadmin 
            password '{macro.COMMON_PASSWD}';'''
        sql_msg = self.sh_primary.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG and
                      self.constant.CREATE_ROLE_SUCCESS_MSG,
                      sql_msg, '执行失败:' + text)
        param_1 = f'local replication {self.user_name}              sha256'
        param_2 = f'host  replication {self.user_name} 127.0.0.1/32 sha256'
        param_3 = f'host  replication {self.user_name} ::1/128      sha256'
        result_1 = self.sh_primary.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_1)
        result_2 = self.sh_primary.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_2)
        result_3 = self.sh_primary.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_3)
        self.assertTrue(result_1 and result_2 and result_3, '执行失败' + text)

        text = "-----step3:非初始用户创建逻辑复制槽，输入错误密码;" \
               "expect:创建失败，合理报错-----"
        self.logger.info(text)
        create_slot_msg = self.sh_primary.exec_pg_recvlogical(
            '--create', 'postgres', self.slot_name_1, self.user_name,
            '-s 2 -W', self.wrong_password)
        self.logger.info(create_slot_msg)
        self.assertIn("FATAL:  Invalid username/password,login denied",
                      create_slot_msg, '执行失败:' + text)

        text = "-----step4:非初始用户使用-W参数创建逻辑复制槽，输入正确密码;" \
               "expect:创建成功-----"
        self.logger.info(text)
        create_slot_msg = self.sh_primary.exec_pg_recvlogical(
            '--create', 'postgres', self.slot_name_1, self.user_name, '-s 2 -W')
        self.logger.info(create_slot_msg)
        text = "-----验证创建逻辑复制槽成功;expect:创建成功-----"
        self.logger.info(text)
        sql_cmd = f"select * from pg_get_replication_slots() " \
            f"where slot_name like '%{self.slot_name_1}%';"
        sql_msg = self.sh_primary.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(self.slot_name_1, sql_msg, '执行失败:' + text)

        text = "-----step5:非初始用户使用-W参数开启逻辑复制槽，输入正确密码;" \
               "expect:开启成功-----"
        self.logger.info(text)
        thread_1 = ComThread(self.sh_primary.exec_pg_recvlogical,
                             args=('--start', 'postgres', self.slot_name_1,
                                   self.user_name,
                                   f'-f - -F 1 -s 2 -W',))
        thread_1.setDaemon(True)
        thread_1.start()
        thread_1.join(10)
        self.logger.info(thread_1.get_result())
        text = "-----验证主机开启逻辑复制槽成功;expect:开启成功-----"
        self.logger.info(text)
        sql_cmd = f"select active from pg_get_replication_slots() " \
            f"where slot_name like '{self.slot_name_1}%';;"
        sql_msg = self.sh_primary.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertEquals('t', sql_msg.splitlines()[-2].strip(),
                          '执行失败:' + text)
        stop_cmd = "ps -ef | grep  pg_recvlogical | grep -v grep | " \
                   "awk '{{print $2}}' | xargs kill -9"
        self.logger.info(stop_cmd)
        primary_result = self.primary_node.sh(stop_cmd).result()
        self.logger.info(primary_result)

        text = "-----step6:非初始用户使用-W参数删除逻辑复制槽，输入正确密码;" \
               "expect:删除成功-----"
        self.logger.info(text)
        drop_slot_msg = self.sh_primary.exec_pg_recvlogical(
            '--drop', 'postgres', self.slot_name_1, self.user_name, '-s 2 -W')
        self.logger.info(drop_slot_msg)
        sql_cmd = f"select * from pg_get_replication_slots() " \
            f"where slot_name like '%{self.slot_name_1}%';"
        sql_msg = self.sh_primary.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn("(0 rows)", sql_msg, '执行失败:' + text)

        text = "-----step7:非初始用户不使用-W参数创建逻辑复制槽，输入正确密码;" \
               "expect:创建成功-----"
        self.logger.info(text)
        create_slot_msg = self.sh_primary.exec_pg_recvlogical(
            '--create', 'postgres', self.slot_name_2, self.user_name, '-s 2 ')
        self.logger.info(create_slot_msg)
        text = "-----验证创建逻辑复制槽成功;expect:创建成功-----"
        self.logger.info(text)
        sql_cmd = f"select * from pg_get_replication_slots() " \
            f"where slot_name like '%{self.slot_name_2}%';"
        sql_msg = self.sh_primary.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn(self.slot_name_2, sql_msg, '执行失败:' + text)

        text = "-----step8:非初始用户不使用-W参数开启逻辑复制槽，输入正确密码;" \
               "expect:开启成功-----"
        self.logger.info(text)
        thread_1 = ComThread(self.sh_primary.exec_pg_recvlogical,
                             args=('--start', 'postgres', self.slot_name_2,
                                   self.user_name,
                                   f'-f - -F 1 -s 2 ',))
        thread_1.setDaemon(True)
        thread_1.start()
        thread_1.join(10)
        self.logger.info(thread_1.get_result())
        text = "-----验证主机开启逻辑复制槽成功;expect:开启成功-----"
        self.logger.info(text)
        sql_cmd = f"select active from pg_get_replication_slots() " \
            f"where slot_name like '{self.slot_name_2}%';;"
        sql_msg = self.sh_primary.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertEquals('t', sql_msg.splitlines()[-2].strip(),
                          '执行失败:' + text)
        stop_cmd = "ps -ef | grep  pg_recvlogical | grep -v grep | " \
                   "awk '{{print $2}}' | xargs kill -9"
        self.logger.info(stop_cmd)
        primary_result = self.primary_node.sh(stop_cmd).result()
        self.logger.info(primary_result)

        text = "-----step9:非初始用户不使用-W参数删除逻辑复制槽，输入正确密码;" \
               "expect:删除成功-----"
        self.logger.info(text)
        drop_slot_msg = self.sh_primary.exec_pg_recvlogical(
            '--drop', 'postgres', self.slot_name_2, self.user_name, '-s 2 ')
        self.logger.info(drop_slot_msg)
        sql_cmd = f"select * from pg_get_replication_slots() " \
            f"where slot_name like '%{self.slot_name_2}%';"
        sql_msg = self.sh_primary.execut_db_sql(sql_cmd, dbname='postgres')
        self.logger.info(sql_msg)
        self.assertIn("(0 rows)", sql_msg, '执行失败:' + text)

    def tearDown(self):
        text_1 = "-----step10:删除用户;expect:删除成功-----"
        self.logger.info(text_1)
        drop_cmd = f'''drop user {self.user_name};'''
        drop_msg = self.sh_primary.execut_db_sql(drop_cmd, dbname='postgres')
        self.logger.info(drop_msg)

        text_2 = "-----step11:恢复参数，恢复pg_hba.conf文件，" \
                 "expect:恢复成功-----"
        self.logger.info(text_2)
        param_1 = f'local replication {self.user_name} '
        param_2 = f'host  replication {self.user_name} 127.0.0.1/32 '
        param_3 = f'host  replication {self.user_name} ::1/128 '
        result_1 = self.sh_primary.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_1)
        result_2 = self.sh_primary.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_2)
        result_3 = self.sh_primary.execute_gsguc(
            'reload', self.constant.GSGUC_SUCCESS_MSG, '', pghba_param=param_3)

        wal_level_current = self.common.show_param(self.config_wal_level)
        if wal_level_current != self.wal_level_default:
            self.sh_primary.execute_gsguc(
                'set', self.constant.GSGUC_SUCCESS_MSG,
                f'{self.config_wal_level}={self.wal_level_default}')
            restart_msg = self.sh_primary.restart_db_cluster()
            self.logger.info(restart_msg)
        show_wal_level = self.common.show_param(self.config_wal_level)

        self.assertIn(self.constant.DROP_ROLE_SUCCESS_MSG,
                      drop_msg, '执行失败' + text_1)
        self.assertTrue(result_1 and result_2 and result_3, '执行失败' + text_2)
        self.assertEquals(show_wal_level, self.wal_level_default,
                          '执行失败' + text_2)
        self.logger.info(f"-----{os.path.basename(__file__)} end-----")
